package org.fb.boot;

import com.alibaba.fastjson.JSONObject;
import io.netty.bootstrap.Bootstrap;
import io.netty.channel.ChannelInitializer;
import io.netty.channel.ChannelOption;
import io.netty.channel.ChannelPipeline;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioSocketChannel;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.ChildData;
import org.apache.curator.framework.recipes.cache.TreeCache;
import org.apache.curator.framework.recipes.cache.TreeCacheListener;
import org.apache.zookeeper.data.Stat;
import org.fb.bean.UserServiceBeanDefinition;
import org.fb.bean.UserServiceBeanFactory;
import org.fb.dto.RpcRequest;
import org.fb.dto.RpcResponse;
import org.fb.handler.UserClientHandler;
import org.fb.service.IUserService;
import org.fb.utils.JSONSerializer;
import org.fb.utils.RpcDecoder;
import org.fb.utils.RpcEncoder;
import org.fb.utils.ZkClient;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.annotation.Resource;
import java.lang.reflect.Proxy;
import java.text.MessageFormat;
import java.util.ArrayList;
import java.util.List;
import java.util.UUID;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * @author fanbo@imoran.net
 * @date 2021/2/23 11:30
 */

@Component
public class RpcClientBoot {
    @Value("${zookeeper.server}")
    private String zookeeperServer;

    @Value(("${zookeeper.sessionTimeoutMs}"))
    private int sessionTimeoutMs;

    @Value("${zookeeper.connectionTimeoutMs}")
    private int connectionTimeoutMs;

    @Value("${zookeeper.maxRetries}")
    private int maxRetries;

    @Value("${zookeeper.baseSleepTimeMs}")
    private int baseSleepTimeMs;

    @Resource(name = "threadPoolExecutor")
    private ThreadPoolExecutor threadPoolExecutor;

    /**
     * 启动初始化步骤，连接zookeeper, 获取服务节点， 连接服务节点， 启动对服务节点状态的监听
     * @throws Exception
     */
    @PostConstruct
    public void getServiceNode() throws Exception {
        ZkClient zkClient = new ZkClient();
        zkClient.init();
        String rootNode = "/services";
        List<String> children = zkClient.getChildren(rootNode);

        connectRpcServer(zkClient, rootNode, children);

        TreeCache treeCache = registerTreeCacheListener(zkClient.getClient(), rootNode, 2, (client, event) -> {
            ChildData childData = event.getData();

            if (childData != null) {
                System.out.println("Path: " + childData.getPath());
                //System.out.println("Stat:" + childData.getStat());
                System.out.println("Data: " + new String(childData.getData()));
            }

            switch (event.getType()) {
                case NODE_ADDED:
                    System.out.println("正在新增子节点：" + childData.getPath());
                    //获取子节点
                    List<String> list = zkClient.getChildren(rootNode);
                    // 重新建立连接
                    connectRpcServer(zkClient, rootNode, list);
                    break;
                case NODE_UPDATED:
                    System.out.println("正在更新节点：" + childData.getPath());
                    break;
                case NODE_REMOVED:
                    //获取子节点
                    list = zkClient.getChildren(rootNode);
                    System.out.println("节点被删除：" + childData.getPath());
                    // 重新建立连接
                    connectRpcServer(zkClient, rootNode, list);
                    break;
                case CONNECTION_LOST:
                    System.out.println("连接丢失");
                    break;
                case CONNECTION_SUSPENDED:
                    System.out.println("连接被挂起");
                    break;
                case CONNECTION_RECONNECTED:
                    System.out.println("恢复连接");
                    break;
            }
        });

    }


    /**
     * 客户端与每一个服务端都建立连接
     *
     * @param children
     */
    private void connectRpcServer(ZkClient zkClient, String rootNode, List<String> children) throws Exception {
        UserServiceBeanFactory.beanList.clear();

        for (String child : children) {
            Stat stat = new Stat();
            String data = new String(zkClient
                    .getClient()
                    .getData()
                    .storingStatIn(stat)
                    .forPath(rootNode + "/" + child));

            System.out.println("节点数据内容： " + data);
            String[] split = data.split(":");
            String host = split[0];
            Integer port = Integer.parseInt(split[1]);


            initClient(host, port, child);
        }
    }

    /**
     * 3.编写方法,初始化客户端  ( 创建连接池  bootStrap  设置bootstrap  连接服务器)
     */
    private void initClient(String host, int port, String serviceName) throws InterruptedException {
        //1) 初始化UserClientHandler
        UserClientHandler userClientHandler = new UserClientHandler();

        //2)创建连接池对象
        NioEventLoopGroup group = new NioEventLoopGroup();

        //3)创建客户端的引导对象
        Bootstrap bootstrap = new Bootstrap();

        //4)配置启动引导对象
        bootstrap.group(group)
                //设置通道为NIO
                .channel(NioSocketChannel.class)
                //设置请求协议为TCP
                .option(ChannelOption.TCP_NODELAY, true)
                //监听channel 并初始化
                .handler(new ChannelInitializer<SocketChannel>() {
                    @Override
                    protected void initChannel(SocketChannel socketChannel) throws Exception {
                        //获取ChannelPipeline
                        ChannelPipeline pipeline = socketChannel.pipeline();
                        //设置编码
                        pipeline.addLast(new RpcEncoder(RpcRequest.class, new JSONSerializer()));
                        pipeline.addLast(new RpcDecoder(RpcResponse.class, new JSONSerializer()));

                        //添加自定义事件处理器
                        pipeline.addLast(userClientHandler);
                    }
                });
        //5)连接服务端
        System.out.println("建立连接： " + host + ":" + port);
        bootstrap.connect(host, port).sync();

        // 创建UserService代理对象
        IUserService userService = (IUserService) createProxy(IUserService.class, userClientHandler);
        UserServiceBeanDefinition beanDefinition = new UserServiceBeanDefinition();
        beanDefinition.setUserService(userService);
        beanDefinition.setServerName(serviceName);

        // 将代理对象统一交给BeanFactory管理
        System.out.println("UserServiceBeanDefinition: " + JSONObject.toJSONString(beanDefinition));
        UserServiceBeanFactory.beanList.add(beanDefinition);
    }

    /**
     * 使用JDK动态代理创建对象
     *
     * @param serviceClass 接口类型，根据哪个接口生成代理对象
     * @return
     */
    private Object createProxy(Class<?> serviceClass, UserClientHandler userClientHandler) {
        return Proxy.newProxyInstance(Thread.currentThread().getContextClassLoader(),
                new Class[]{serviceClass}, (proxy, method, args) -> {

                    RpcRequest request = new RpcRequest();
                    String requestId = UUID.randomUUID().toString();

                    String className = method.getDeclaringClass().getName();
                    String methodName = method.getName();

                    Class<?>[] parameterTypes = method.getParameterTypes();

                    request.setRequestId(requestId);
                    request.setClassName(className);
                    request.setMethodName(methodName);
                    request.setParameterTypes(parameterTypes);
                    request.setParameters(args);

                    // 设置参数
                    userClientHandler.setParam(request);

                    // 使用线程池，开启一个线程写操作，并返回结果
                    Object result = threadPoolExecutor.submit(userClientHandler).get();
                    // 将结果return
                    return result;
                });
    }


    /**
     * 注册目录监听器
     * TreeCache：综合NodeCache和PathChildrenCahce的特性，可以对整个目录进行监听，同时还可以设置监听深度
     *
     * @param nodePath 节点路径
     * @param maxDepth 自定义监控深度
     * @param listener 监控事件的回调接口
     * @return org.apache.curator.framework.recipes.cache.TreeCache
     */
    public TreeCache registerTreeCacheListener(CuratorFramework client, String nodePath, int maxDepth, TreeCacheListener listener) {
        try {
            //1. 创建一个TreeCache
            TreeCache treeCache = TreeCache.newBuilder(client, nodePath)
                    .setCacheData(true)
                    .setMaxDepth(maxDepth)
                    .build();

            //2. 添加目录监听器
            treeCache.getListenable().addListener(listener);

            //3. 启动监听器
            treeCache.start();

            //4. 返回TreeCache
            return treeCache;
        } catch (Exception e) {
            System.out.println(MessageFormat.format("注册目录监听器出现异常,nodePath:{0},maxDepth:{1}", nodePath));
        }
        return null;
    }
}
